package com.dtp.core;

import com.dtp.common.entity.DtpExecutorProps;
import com.dtp.common.entity.TpMainFields;
import com.dtp.common.ex.DtpException;
import com.dtp.common.properties.DtpProperties;
import com.dtp.common.queue.MemorySafeLinkedBlockingQueue;
import com.dtp.common.queue.VariableLinkedBlockingQueue;
import com.dtp.core.convert.ExecutorConverter;
import com.dtp.core.notify.manager.NoticeManager;
import com.dtp.core.reject.RejectHandlerGetter;
import com.dtp.core.support.ExecutorWrapper;
import com.dtp.core.support.wrapper.TaskWrapper;
import com.dtp.core.support.wrapper.TaskWrappers;
import com.dtp.core.thread.DtpExecutor;
import com.dtp.core.thread.ExecutorAdapter;
import com.github.dadiyang.equator.Equator;
import com.github.dadiyang.equator.FieldInfo;
import com.github.dadiyang.equator.GetterBaseEquator;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

import static com.dtp.common.constant.DynamicTpConst.M_1;
import static com.dtp.common.constant.DynamicTpConst.PROPERTIES_CHANGE_SHOW_STYLE;
import static com.dtp.core.notify.manager.NotifyHelper.updateNotifyInfo;
import static java.util.stream.Collectors.toList;

/**
 * 核心注册表，它保存所有已注册的动态线程池
 *
 * @author yanhom
 * @since 1.0.0
 *
 * @remark muse
 **/
@Slf4j
public class DtpRegistry implements ApplicationRunner, Ordered {
    private static final Map<String, DtpExecutor> DTP_REGISTRY = new ConcurrentHashMap<>();         // 维护所有自动注册和手动注册的DtpExecutors
    private static final Map<String, ExecutorWrapper> COMMON_REGISTRY = new ConcurrentHashMap<>();  // 维护所有自动注册和手动注册的JUC的线程池
    private static final Equator EQUATOR = new GetterBaseEquator(); // 属性比对器（一个用于比较两个对象的属性是否相等，并且可以获取所有不相等的属性的比对器）
    private static DtpProperties dtpProperties;                     // 由配置中心维护的主要属性

    /** 获得所有DtpExecutor的名称列表 */
    public static List<String> listAllDtpNames() {return Lists.newArrayList(DTP_REGISTRY.keySet());}

    /** 获得所有JUC线程池执行器的名称列表 */
    public static List<String> listAllCommonNames() {return Lists.newArrayList(COMMON_REGISTRY.keySet());}

    /** 注册一个DtpExecutor */
    public static void registerDtp(DtpExecutor executor, String source) {
        DTP_REGISTRY.putIfAbsent(executor.getThreadPoolName(), executor);
    }

    /** 注册一个普通线程池ThreadPoolExecutor */
    public static void registerCommon(ExecutorWrapper wrapper, String source) {
        COMMON_REGISTRY.putIfAbsent(wrapper.getThreadPoolName(), wrapper);
    }

    /** 通过线程池名称，获得一个动态线程池DtpExecutor */
    public static DtpExecutor getDtpExecutor(final String name) {
        val executor = DTP_REGISTRY.get(name);
        if (Objects.isNull(executor)) throw new DtpException("Cannot find a specified dtpExecutor, name: " + name);

        return executor;
    }

    /** 通过线程池名称，获得一个普通线程池ThreadPoolExecutor */
    public static ExecutorWrapper getCommonExecutor(final String name) {
        val executor = COMMON_REGISTRY.get(name);
        if (Objects.isNull(executor)) throw new DtpException("Cannot find a specified commonExecutor, name: " + name);

        return executor;
    }

    /** 当监听配置改变时刷新操作 */
    public static void refresh(DtpProperties dtpProperties) {
        if (Objects.isNull(dtpProperties) || CollectionUtils.isEmpty(dtpProperties.getExecutors()))
            return;

        dtpProperties.getExecutors().forEach(x -> {
            if (StringUtils.isBlank(x.getThreadPoolName())) return;

            // 首先，在DTP_REGISTRY中查找它，如果存在dtp执行器，则执行刷新操作
            val dtpExecutor = DTP_REGISTRY.get(x.getThreadPoolName());
            if (Objects.nonNull(dtpExecutor)) {
                refresh(ExecutorWrapper.of(dtpExecutor), x);
                return;
            }

            // 其次，在COMMON_REGISTRY中查找它，如果存在普通执行器，则执行刷新操作
            val executorWrapper = COMMON_REGISTRY.get(x.getThreadPoolName());
            if (Objects.nonNull(executorWrapper)) {
                refresh(executorWrapper, x);
                return;
            }

            log.warn("DynamicTp refresh, cannot find specified executor, name: {}.", x.getThreadPoolName());
        });
    }

    /** 当监听配置改变时刷新操作 */
    private static void refresh(ExecutorWrapper executorWrapper, DtpExecutorProps props) {
        if (props.coreParamIsInValid()) return; // 当执行动态线程池刷新操作的时候，发现存在无效参数

        // 获得刷新前（oldFields）和刷新后（newFields）的线程池主要配置信息
        TpMainFields oldFields = ExecutorConverter.convert(executorWrapper);
        doRefresh(executorWrapper, props);
        TpMainFields newFields = ExecutorConverter.convert(executorWrapper);

        // 如果配置信息没有变化，则直接返回
        if (oldFields.equals(newFields)) {
            log.debug("DynamicTp refresh, main properties of [{}] have not changed.", executorWrapper.getThreadPoolName());
            return;
        }

        // 获取配置之间的差异项，执行异步通知
        List<String> diffKeys = EQUATOR.getDiffFields(oldFields, newFields).stream().map(FieldInfo::getFieldName).collect(toList());
        NoticeManager.doNoticeAsync(executorWrapper, oldFields, diffKeys);

        log.info("DynamicTp refresh, name: [{}], changed keys: {}, corePoolSize: [{}], maxPoolSize: [{}]," +
                        " queueType: [{}], queueCapacity: [{}], keepAliveTime: [{}], rejectedType: [{}]," +
                        " allowsCoreThreadTimeOut: [{}]", executorWrapper.getThreadPoolName(), diffKeys,
                String.format(PROPERTIES_CHANGE_SHOW_STYLE, oldFields.getCorePoolSize(), newFields.getCorePoolSize()),
                String.format(PROPERTIES_CHANGE_SHOW_STYLE, oldFields.getMaxPoolSize(), newFields.getMaxPoolSize()),
                String.format(PROPERTIES_CHANGE_SHOW_STYLE, oldFields.getQueueType(), newFields.getQueueType()),
                String.format(PROPERTIES_CHANGE_SHOW_STYLE, oldFields.getQueueCapacity(), newFields.getQueueCapacity()),
                String.format("%ss => %ss", oldFields.getKeepAliveTime(), newFields.getKeepAliveTime()),
                String.format(PROPERTIES_CHANGE_SHOW_STYLE, oldFields.getRejectType(), newFields.getRejectType()),
                String.format(PROPERTIES_CHANGE_SHOW_STYLE, oldFields.isAllowCoreThreadTimeOut(),
                        newFields.isAllowCoreThreadTimeOut()));
    }

    /** 真正执行刷新操作的地方 */
    private static void doRefresh(ExecutorWrapper executorWrapper, DtpExecutorProps props) {
        ExecutorAdapter<?> executor = executorWrapper.getExecutor();

        // 更新线程池中的【corePoolSize】和【maximumPoolSize】，以props的数量为主
        doRefreshPoolSize(executor, props);

        // 更新线程池中的【keepAliveTime】和【allowCoreThreadTimeOut】，以props的数量为主
        if (!Objects.equals(executor.getKeepAliveTime(props.getUnit()), props.getKeepAliveTime()))
            executor.setKeepAliveTime(props.getKeepAliveTime(), props.getUnit());
        if (!Objects.equals(executor.allowsCoreThreadTimeOut(), props.isAllowCoreThreadTimeOut()))
            executor.allowCoreThreadTimeOut(props.isAllowCoreThreadTimeOut());

        // 更新阻塞队列中的【maxFreeMemory】或【capacity】，以props的数量为主
        updateQueueProps(executor, props);

        if (executor instanceof DtpExecutor) {
            doRefreshDtp((DtpExecutor) executor, props); // 更新dtp执行器中的参数，以props为主
            return;
        }

        // 更新普通执行器中的参数，以配置为主
        doRefreshCommon(executorWrapper, props);
    }

    /** 更新【CorePoolSize】和【MaximumPoolSize】，以配置的数量为主 */
    private static void doRefreshPoolSize(ExecutorAdapter<?> executor, DtpExecutorProps props) {
        if (props.getMaximumPoolSize() < executor.getMaximumPoolSize()) {
            if (!Objects.equals(executor.getCorePoolSize(), props.getCorePoolSize()))
                executor.setCorePoolSize(props.getCorePoolSize());
            if (!Objects.equals(executor.getMaximumPoolSize(), props.getMaximumPoolSize()))
                executor.setMaximumPoolSize(props.getMaximumPoolSize());
            return;
        }

        if (!Objects.equals(executor.getMaximumPoolSize(), props.getMaximumPoolSize()))
            executor.setMaximumPoolSize(props.getMaximumPoolSize());
        if (!Objects.equals(executor.getCorePoolSize(), props.getCorePoolSize()))
            executor.setCorePoolSize(props.getCorePoolSize());
    }

    /** 更新阻塞队列中的【maxFreeMemory】和【capacity】，以props的数量为主 */
    private static void updateQueueProps(ExecutorAdapter<?> executor, DtpExecutorProps props) {
        val blockingQueue = executor.getQueue();
        if (blockingQueue instanceof MemorySafeLinkedBlockingQueue)
            ((MemorySafeLinkedBlockingQueue<Runnable>) blockingQueue).setMaxFreeMemory(props.getMaxFreeMemory() * M_1);

        if (!(blockingQueue instanceof VariableLinkedBlockingQueue)) {
            log.warn("DynamicTp refresh, the blockingqueue capacity cannot be reset, poolName: {}, queueType {}",
                    props.getThreadPoolName(), blockingQueue.getClass().getSimpleName());
            return;
        }

        int capacity = blockingQueue.size() + blockingQueue.remainingCapacity();
        if (!Objects.equals(capacity, props.getQueueCapacity()))
            ((VariableLinkedBlockingQueue<Runnable>) blockingQueue).setCapacity(props.getQueueCapacity());
    }

    /** 更新dtp执行器中的参数，以配置为主 */
    private static void doRefreshDtp(DtpExecutor executor, DtpExecutorProps props) {
        if (StringUtils.isNotBlank(props.getThreadPoolAliasName()))
            executor.setThreadPoolAliasName(props.getThreadPoolAliasName());

        // update reject handler
        if (!Objects.equals(executor.getRejectHandlerName(), props.getRejectedHandlerType())) {
            executor.setRejectedExecutionHandler(RejectHandlerGetter.getProxy(props.getRejectedHandlerType())); // 获得代理对象
            executor.setRejectHandlerName(props.getRejectedHandlerType());
        }
        executor.setWaitForTasksToCompleteOnShutdown(props.isWaitForTasksToCompleteOnShutdown());
        executor.setAwaitTerminationSeconds(props.getAwaitTerminationSeconds());
        executor.setPreStartAllCoreThreads(props.isPreStartAllCoreThreads());
        executor.setRunTimeout(props.getRunTimeout());
        executor.setQueueTimeout(props.getQueueTimeout());
        List<TaskWrapper> taskWrappers = TaskWrappers.getInstance().getByNames(props.getTaskWrapperNames());
        executor.setTaskWrappers(taskWrappers);

        // update notify related
        updateNotifyInfo(executor, props, dtpProperties.getPlatforms());
    }

    /** 更新普通执行器中的参数，以配置为主 */
    private static void doRefreshCommon(ExecutorWrapper executorWrapper, DtpExecutorProps props) {
        if (StringUtils.isNotBlank(props.getThreadPoolAliasName()))
            executorWrapper.setThreadPoolAliasName(props.getThreadPoolAliasName());

        ExecutorAdapter<?> executor = executorWrapper.getExecutor();
        // update reject handler
        String currentRejectHandlerName = executor.getRejectHandlerName();
        if (!Objects.equals(currentRejectHandlerName, props.getRejectedHandlerType())) {
            val rejectHandler = RejectHandlerGetter.buildRejectedHandler(props.getRejectedHandlerType());
            executor.setRejectedExecutionHandler(rejectHandler);
        }

        // update notify related
        updateNotifyInfo(executorWrapper, props, dtpProperties.getPlatforms());
    }

    @Override
    public void run(ApplicationArguments args) {
        // 远端配置（配置中心）的线程池名称列表——remoteExecutors
        Set<String> remoteExecutors = Collections.emptySet();
        if (CollectionUtils.isNotEmpty(dtpProperties.getExecutors()))
            remoteExecutors = dtpProperties.getExecutors().stream().map(DtpExecutorProps::getThreadPoolName)
                    .collect(Collectors.toSet());

        // 聚合注册过的的所有线程池——registeredExecutors
        val registeredExecutors = Sets.newHashSet(DTP_REGISTRY.keySet());
        registeredExecutors.addAll(COMMON_REGISTRY.keySet());

        // 寻找本地配置的线程池——localExecutors
        val localExecutors = CollectionUtils.subtract(registeredExecutors, remoteExecutors);

        // 【localExecutors】+【remoteExecutors】=【registeredExecutors】
        log.info("DtpRegistry has been initialized, remote executors: {}, local executors: {}", remoteExecutors, localExecutors);
    }

    @Autowired
    public void setDtpProperties(DtpProperties dtpProperties) {DtpRegistry.dtpProperties = dtpProperties;}

    @Override
    public int getOrder() {return Ordered.HIGHEST_PRECEDENCE + 1;}
}
